package cas.ihep.spark.rdd;

import cas.ihep.spark.util.IterationClientListener;
import cas.ihep.spark.util.NamedResourceManager;
import cas.ihep.spark.util.function.Function2;
import cas.ihep.spark.util.function.Function3;
import cas.ihep.spark.util.function.Function4;
import scala.reflect.ClassManifestFactory;
import scala.reflect.ClassTag;

import java.io.Serializable;
import java.net.SocketAddress;
import java.util.Map;

class BaseBroadcastData<Send extends Serializable, Receive extends Serializable> implements Serializable{
    SocketAddress serverAddr;
    Map<String,String> properties;
    ClassTag<Send> sendClassTag;
    Receive zero;
    Function2<Receive, Receive, Receive> aggregateFunc;
    BaseBroadcastData(SocketAddress addr, Class<Send> pcls, Receive z, Function2<Receive, Receive, Receive> agg){
        serverAddr=addr;
        sendClassTag =ClassManifestFactory.classType(pcls);
        zero=z;
        aggregateFunc=agg;
    }
    static final String CLIENT_KEY="ImtheBEST!";
}

class IterativeBroadcastData<T, Send extends Serializable, Receive extends Serializable> extends BaseBroadcastData<Send, Receive>{

    Function3<T, Send, NamedResourceManager, Receive> clientFunc;
    IterativeBroadcastData(SocketAddress addr, Class<Send> pcls, Receive z, Function2<Receive, Receive, Receive> agg, Function3<T, Send,NamedResourceManager, Receive> client) {
        super(addr, pcls, z, agg);
        clientFunc=client;
    }
}

class IterativeBroadcastDataKV<K,V, Send extends Serializable, Receive extends Serializable> extends BaseBroadcastData<Send, Receive> {

    Function4<K,V, Send, NamedResourceManager, Receive> clientFunc;

    IterativeBroadcastDataKV(SocketAddress addr, Class<Send> pcls, Receive z, Function2<Receive, Receive, Receive> agg, Function4<K,V, Send,NamedResourceManager, Receive> client) {
        super(addr, pcls, z, agg);
        clientFunc=client;
    }
}

class IterativeBroadcastDatav2<T, Send extends Serializable, Receive extends Serializable> extends BaseBroadcastData<Send, Receive>{
    IterationClientListener<T, Send, Receive> clientListener;
    IterativeBroadcastDatav2(SocketAddress addr, Class<Send> pcls, Receive z, Function2<Receive, Receive, Receive> agg, IterationClientListener<T, Send, Receive> cRun) {
        super(addr, pcls, z, agg);
        clientListener =cRun;
    }
}

